package com.iacrqq.sedaf.queue;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.iacrqq.sedaf.config.Configuration;
import com.iacrqq.sedaf.queue.exception.QueueFullException;

/**
 * 
 * @author sihai
 * 
 * @param <T>
 */
public class FiniteQueue<T> extends AbstractQueue<T>
{
	private static final Log logger = LogFactory.getLog(FiniteQueue.class);

	private String name;
	private volatile int capacity;
	private BlockingQueue<T> store;

	private QueueRuntimeStatisticsCollector<T> queueRuntimeStatisticsCollector;

	private Object _lock_;

	@Override
	public String getName()
	{
		return name;
	}

	@Override
	public void setName(String name)
	{
		this.name = name;
	}

	@Override
	public void setCapacity(int capacity)
	{
		this.capacity = capacity;
	}

	@Override
	public void init(Configuration config) throws Exception
	{
		this.store = new LinkedBlockingQueue<T>(capacity);
		_lock_ = new Object();
		queueRuntimeStatisticsCollector = new QueueRuntimeStatisticsCollector<T>();
		logger.debug(new StringBuilder("Queue: ").append(name).append(" capacity=").append(capacity)
				.append(", initilized."));
	}

	@Override
	public int size()
	{
		return store.size();
	}

	@Override
	public void enqueue(List<T> elementList) throws QueueFullException
	{
		synchronized (_lock_)
		{
			if (store.size() + elementList.size() > capacity)
			{
				queueRuntimeStatisticsCollector.full();
				throw new QueueFullException(new StringBuilder("Queue:").append(name).append(" is full.").toString());
			}
			
			//logger.info(new StringBuilder(name).append(" enqueue").append(elementList.size()).append(" elements").toString());
			store.addAll(elementList);
		}

		queueRuntimeStatisticsCollector.enqueue(elementList);
	}

	@Override
	public void enqueue(T element) throws QueueFullException
	{
		synchronized (_lock_)
		{
			// System.out.println(new
			// StringBuilder("Queue:").append(name).append(", size=").append(store.size()).append(", capacity=").append(capacity).toString());
			if (store.size() + 1 > capacity)
			{
				queueRuntimeStatisticsCollector.full();
				throw new QueueFullException(new StringBuilder("Queue:").append(name).append(" is full.").toString());
			}
			
			//logger.info(new StringBuilder(name).append(" enqueue one element"));
			store.add(element);
		}

		queueRuntimeStatisticsCollector.enqueue(element);
	}

	@Override
	public T blockingDequeue()
	{
		try
		{
			T element = store.take();
			queueRuntimeStatisticsCollector.dequeue(element);
			
			return element;
		}
		catch (InterruptedException e)
		{
			logger.error(e);
			Thread.interrupted();
		}
		
		queueRuntimeStatisticsCollector.emtpy();
		return null;
	}

	@Override
	public T blockingDequeue(long timeout)
	{
		try
		{
			T element = store.poll(timeout, TimeUnit.MILLISECONDS);

			if (element != null)
			{
				queueRuntimeStatisticsCollector.dequeue(element);
			}
			else
			{
				queueRuntimeStatisticsCollector.emtpy();
			}
			
			return element;
		}
		catch (InterruptedException e)
		{
			logger.error(e);
			Thread.interrupted();
		}

		return null;
	}

	@Override
	public T dequeue()
	{
		T element = store.poll();
		if (element != null)
		{
			queueRuntimeStatisticsCollector.dequeue(element);
			//logger.info(new StringBuilder(name).append(" dequeue").append(" one elements").toString());
		}
		
		return element;
	}

	@Override
	public List<T> dequeue(int num)
	{
		List<T> elementList = new LinkedList<T>();
		store.drainTo(elementList, num);
		queueRuntimeStatisticsCollector.dequeue(elementList);
		return elementList;
	}

	@Override
	public List<T> dequeueAll()
	{
		List<T> elementList = new LinkedList<T>();
		store.drainTo(elementList);
		queueRuntimeStatisticsCollector.dequeue(elementList);
		return elementList;
	}

	@Override
	public void destroy() throws Exception
	{
		store.clear();
	}

	@Override
	public QueueRuntimeStatistics getQueueRuntimeStatistics()
	{
		queueRuntimeStatisticsCollector.setSize(size());
		queueRuntimeStatisticsCollector.setCapacity(capacity);
		return queueRuntimeStatisticsCollector.snapshot();
	}
}
